package com.boarsoft.flow.core;

import java.io.IOException;
import java.io.ObjectOutput;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import com.boarsoft.common.Util;
import com.boarsoft.common.util.InetUtil;
import com.boarsoft.common.util.JsonUtil;
import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.flow.core.bean.BaseNode;
import com.boarsoft.flow.core.bean.BeginTxNode;
import com.boarsoft.flow.core.bean.CatchNode;
import com.boarsoft.flow.core.bean.CommitTxNode;
import com.boarsoft.flow.core.bean.FinallyNode;
import com.boarsoft.flow.core.bean.ForkNode;
import com.boarsoft.flow.core.bean.InputNode;
import com.boarsoft.flow.core.bean.JoinNode;
import com.boarsoft.flow.core.bean.JudgeNode;
import com.boarsoft.flow.core.bean.ProcessNode;
import com.boarsoft.flow.core.bean.SubflowNode;
import com.boarsoft.flow.core.bean.SuspendNode;
import com.boarsoft.flow.core.bean.SwitchNode;
import com.boarsoft.flow.core.bean.TransactionHandler;
import com.boarsoft.flow.core.bean.TryNode;
import com.boarsoft.flow.core.bean.UiBean;
import com.boarsoft.flow.core.bean.WrapHandler;
import com.boarsoft.flow.core.bean.YrtNode;
import com.boarsoft.flow.event.SimpleFlowListener;
import com.boarsoft.flow.event.SimpleFlowListenerImpl;
import com.boarsoft.flow.store.EmptyFlowPersisterImpl;
import com.boarsoft.flow.store.SimpleFlowPersister;
import com.boarsoft.flow.trace.FlowTrace;
import com.boarsoft.flow.trace.SimpleFlowTrace;
import com.boarsoft.flow.trace.SimpleFlowTracker;
import com.boarsoft.flow.trace.impl.SimpleFlowTrackerImpl;
import com.boarsoft.serialize.JavaSerializer;
import com.boarsoft.serialize.ObjectSerializer;
import com.boarsoft.soagov.config.SvcConfig;
import com.boarsoft.soagov.spy.SvcSpy;

/**
 * 流程引擎核心类
 * 
 * @author Mac_J
 *
 */
public class SimpleFlowEngineImpl implements Runnable, SimpleFlowEngine, ApplicationContextAware {
	private static final Logger log = LoggerFactory.getLogger(SimpleFlowEngineImpl.class);

	/** 用于控制超时时间 */
	protected Map<String, Map<String, SimpleFlow>> timeMap = new ConcurrentHashMap<String, Map<String, SimpleFlow>>();
	/** */
	protected ApplicationContext applicationContext;
	/** */
	protected ExecutorService threadPool;

	/** */
	protected SimpleFlowPersister persister = new EmptyFlowPersisterImpl();
	/** */
	protected SimpleFlowListener listener = new SimpleFlowListenerImpl();
	/** */
	protected SimpleFlowTracker tracker = new SimpleFlowTrackerImpl();

	/** 与RPC共享的定时器 */
	protected ScheduledExecutorService scheduler;

	/** 所属系统 */
	protected String catalog;
	/** */
	protected SvcSpy svcSpy;
	/** */
	protected Map<String, Map<String, BaseNode>> nodeMap = new ConcurrentHashMap<>();
	/** 流程模板缓存 */
	protected Map<String, SimpleFlow> defsMap = new ConcurrentHashMap<>();
	/** */
	public static ObjectSerializer serializer = new JavaSerializer();
	/** 调用链采集总开关 */
	protected boolean tracing;
	/** */
	protected ObjectOutput messageSender;

	protected final SimpleFlowEngineImpl _this = this;

	static {
		ServiceLoader<ObjectSerializer> sl = ServiceLoader.load(ObjectSerializer.class);
		Iterator<ObjectSerializer> it = sl.iterator();
		if (it.hasNext()) {
			serializer = it.next();
		}
	}

	@PreDestroy
	public void destroy() {
		log.warn("{} is shutting down", this.getClass().getName());
		scheduler.shutdown();
	}

	@Override
	public void watch() {
		tracing = true;
	}

	@Override
	public void watch(String flowId, int ratio) {
		this.getFlow(flowId).setTraceRatio(ratio);
	}

	@Override
	public void ignore() {
		tracing = false;
	}

	@Override
	public void ignore(String flowId) {
		tracing = false;
	}

	@Override
	public boolean exists(String flowId) {
		return applicationContext.containsBean(flowId);
	}

	@PostConstruct
	public void init() {
		if (Util.strIsEmpty(catalog)) {
			throw new IllegalStateException("Catalog can not be empty");
		}
		scheduler.scheduleWithFixedDelay(this, 1000L, 1000L, TimeUnit.MILLISECONDS);
		defsMap.putAll(applicationContext.getBeansOfType(SimpleFlow.class));
	}

	@Override
	public void run() {
		long now = System.currentTimeMillis();
		// log.debug("Check working queue for flows");
		for (Entry<String, Map<String, SimpleFlow>> en1 : timeMap.entrySet()) {
			Map<String, SimpleFlow> m = en1.getValue();
			for (Entry<String, SimpleFlow> en2 : m.entrySet()) {
				String id = en2.getKey();
				SimpleFlow f = en2.getValue();
				// FLAG_END || FLAG_STOP || FLAG_ERROR || FLAG_TIMEOUT
				if (f.getFlag() != SimpleFlow.FLAG_GO && f.getFlag() != SimpleFlow.FLAG_SUSPEND) {
					m.remove(id);
					continue;
				}
				// SimpleFlow.FLAG_GO || SimpleFlow.FLAG_SUSPEND
				if (f.getDeadline() > now) {
					continue;
				}
				log.warn("Flow {} timeout after {}ms", f, now - f.getStartTime());
				m.remove(id);
				this.timeout(f);
			}
		}
	}

	@Override
	public boolean stop(SimpleFlow flow) {
		return this.stop(flow, SimpleFlow.FLAG_STOP);
	}

	public boolean timeout(SimpleFlow flow) {
		log.warn("Flow {} timeout", flow);
		// synchronized (flow) {
		flow.setFlag(SimpleFlow.FLAG_TIMEOUT);
		// }
		return true;
	}

	public boolean stop(SimpleFlow flow, short flag) {
		log.warn("Stopping flow {}", flow);
		synchronized (flow) {
			flow.setFlag(flag);
			try {
				flow.wait(flow.getTimeout());
			} catch (InterruptedException e) {
				log.error("Be interrupted while waiting flow {} stop", flow, e);
				return false;
			}
		}
		return true;
	}

	@Override
	public SimpleFlow getFlow(String flowId, String exeId) {
		SimpleFlow flow = this.persister.load(flowId, exeId);
		if (flow == null) {
			throw new IllegalStateException(String.format(//
					"Can not reload flow %s/%s", flowId, exeId));
		}
		return flow;
	}

	protected void removeFlow(SimpleFlow flow) {
		this.persister.delete(flow);
		// if (mutex) {
		// synchronized (flowId) {
		// flowId.notify();
		// }
		// }
	}

	protected void time(SimpleFlow flow) {
		flow.setStartTime(System.currentTimeMillis());
		// timeout<=0表示不处理超时
		if (flow.getTimeout() <= 0L) {
			return;
		}
		// submit过来的任务，不需要计数，startTime有被设值
		if (flow.getStartTime() > 0L) {
			return;
		}
		Map<String, SimpleFlow> tm = null;
		tm = timeMap.get(flow.getFlowId());
		if (tm == null) {
			synchronized (timeMap) {
				tm = timeMap.get(flow.getFlowId());
				if (tm == null) {
					tm = new ConcurrentHashMap<String, SimpleFlow>();
					timeMap.put(flow.getFlowId(), tm);
				}
			}
		}
		if (flow.isMutex() && !tm.isEmpty()) {
			throw new RejectedExecutionException(String.format(//
					"Another %s flow is running", flow.toString()));
		}
		flow.setDeadline(flow.getStartTime() + flow.getTimeout());
		tm.put(flow.getId(), flow);
	}

	@Override
	public SimpleFlow start(String flowId, Object data) throws Throwable {
		return this.start(flowId, null, data, null);
	}

	@Override
	public SimpleFlow start(String flowId, Object data, String entry) throws Throwable {
		return this.start(flowId, null, data, entry);
	}

	@Override
	public SimpleFlow create(String flowId) {
		return this.getFlow(flowId).newInstance();
	}

	@Override
	public SimpleFlow getFlow(String flowId) {
		SimpleFlow tf = defsMap.get(flowId);
		if (tf == null) {
			synchronized (defsMap) {
				tf = defsMap.get(flowId);
				if (tf == null) {
					tf = applicationContext.getBean(flowId, SimpleFlow.class);
					defsMap.put(flowId, tf);
				}
			}
		}
		return tf;
	}

	@Override
	public SimpleFlow start(String flowId, String id, Object data, String entry) throws Throwable {
		SimpleFlow f = this.create(flowId);
		f.setId(id);
		f.setData(data);
		return this.start(f, entry);
	}

	@Override
	public SimpleFlow start(SimpleFlow flow, String entry) throws Throwable {
		if (Util.strIsNotEmpty(entry)) {
			flow.setEntry(entry);
		}
		return this.start(flow);
	}

	@Override
	public SimpleFlow start(SimpleFlow flow) throws Throwable {
		// int cr = (svcSpy == null ? 0 : svcSpy.check(flow));
		// if (cr != 0) {
		// throw new RejectedExecutionException(String.valueOf(cr));
		// }
		flow.setAddr(InetUtil.getAddr());
		flow.setStatus(SimpleFlow.STATUS_EXECUTING);
		listener.onBegin(flow);
		if (Util.strIsEmpty(flow.getId())) {
			flow.setId(RandomUtil.genUUID());
		}
		// 检查互斥并开始计时
		this.time(flow);
		persister.save(flow);
		this.mark(flow); // 标记线程
		log.info("Start flow {} from {} with deadline {}", flow, flow.getEntry(), //
				Util.date2str(new Date(flow.getDeadline()), Util.STDDTMF));
		try {
			Object dd = this.go(flow, flow.getEntry(), null, null, flow.getTrace());
			flow.setData(dd);
			return flow;
		} catch (Throwable e) {
			log.error("Error on do flow {}", flow, e);
			flow.setError(e);
			flow.setFlag(SimpleFlow.FLAG_ERROR);
			// 此处不调listener的onError，在onFinally中调用
			return flow;
		} finally {
			this.onFinally(flow);
		}
	}

	protected void onFinally(SimpleFlow flow) throws Throwable {
		try {
			long time = System.currentTimeMillis();
			flow.setEndTime(time);
			time -= flow.getStartTime();
			switch (flow.getFlag()) {
			case SimpleFlow.FLAG_ERROR:
				log.info("Flow {} hit error after run {}ms", flow, time);
				flow.setStatus(SimpleFlow.STATUS_FALED);
				listener.onError(flow); // 允许onError事件修改flow的状态
				persister.update(flow); // 更新状态
				if (flow.getStatus() == SimpleFlow.STATUS_FALED) {
					this.reverse(flow); // 尝试冲正（如果有配的话）
				}
				break;
			case SimpleFlow.FLAG_SUSPEND:
				log.info("Flow {} suspend after run {}ms", flow, time);
				flow.setStatus(SimpleFlow.STATUS_SUSPEND);
				listener.onSuspend(flow);
				if (flow.getStatus() == SimpleFlow.STATUS_SUSPEND) {
					this.persister.update(flow);
					// TODO 根据timeout设置，确定登记到哪一个计时用的Map?
				}
				persister.update(flow); // 更新状态
				break;
			case SimpleFlow.FLAG_STOP:
				log.info("Flow {} stopped after run {}ms", flow, time);
				flow.setStatus(SimpleFlow.STATUS_STOPPED);
				synchronized (flow) {
					flow.notifyAll(); // 通知stop方法（目前只有stop方法会wait）
				}
				listener.onStop(flow);
				persister.update(flow); // 更新状态
				break;
			case SimpleFlow.FLAG_END:
				log.info("Flow {} complete after run {}ms", flow, time);
				flow.setStatus(SimpleFlow.STATUS_SUCCESS);
				listener.onEnd(flow);
				persister.update(flow); // 更新状态
				break;
			case SimpleFlow.FLAG_TIMEOUT:
				log.error("Flow {} timout after run {}ms", flow, time);
				flow.setError(new TimeoutException(//
						String.format("Flow %s timeout", flow.toString())));
				// 为了避免正交易超时后触发联机冲正与批量冲正重复，约定：正交易超时不触发联机冲正！
				// 超时后去触发联机冲正，有较大机率也发生超时，且不能确定（外联超时）结果
				// this.reverse(flow);
				listener.onTimeout(flow);
				persister.update(flow); // 更新状态
				break;
			default:
				log.error("Flow {} be stopped with invalid flag {}", flow, flow.getFlag());
				flow.setError(new IllegalStateException(//
						String.format("Flow %s be stopped with invalid flag %d", flow, flow.getFlag())));
				flow.setStatus(SimpleFlow.STATUS_FALED);
				listener.onError(flow); // 允许onError事件修改flow的状态
				persister.update(flow); // 更新状态
				break;
			}
		} finally {
			listener.onFinally(flow); // 用于简化和集中的收尾处理
			this.unmark(flow);
		}
		// 发送流程结束事件
		this.sendTrace(flow);
		// 有异常则抛出
		if (flow.getError() != null) {
			throw flow.getError();
		}
	}

	protected void sendTrace(SimpleFlow flow) {
		try {
			// 全局跟踪开关已打开，当前流程有配置采样率，非子流程
			if (tracing && flow.isTracing() && !flow.hasParent()) {
				String k = new StringBuilder().append(catalog)//
						.append(".").append(flow.getFlowId()).toString();
				String t = flow.getTraceTarget();
				if (Util.strIsEmpty(t)) {
					t = k;
				}
				Map<String, Object> m = new HashMap<String, Object>();
				m.put("id", flow.getId());
				m.put("target", t);
				m.put("key", k);
				m.put("content", JsonUtil.toJSONString(flow.getTrace()));
				messageSender.writeObject(m);
			}
		} catch (Exception e) {
			log.warn("Error on send trace of flow {}", flow, e);
		}
	}

	protected void reverse(final SimpleFlow flow) throws Throwable {
		short rt = flow.getRevType();
		if (rt == SimpleFlow.REV_TYPE_IGNORE) {
			return;
		}
		String rfc = flow.getRevCode();
		if (Util.strIsEmpty(rfc)) {
			log.warn("Reverse code of flow {} is empty", flow.toString());
			return;
		}
		BaseNode bn = this.getNode(flow.getFlowId(), flow.getStep());
		String re = bn.getRevCode();
		// 不配revCode则默认从头开始执行
		if (isEndStep(re)) {
			log.warn("Reverse code of flow node {} is {}", bn.getId(), re);
			return;
		}
		//
		if (rt == SimpleFlow.REV_TYPE_BATCH) {
			flow.setRevEntry(re);
			// 正反交易不一定在同一个节点上，也不一定存在远程引用，所以不能创建反易记录
			persister.update(flow);
			return;
		}
		//
		if (!applicationContext.containsBean(rfc)) {
			flow.setError(new IllegalStateException(//
					String.format("Flow %s not found", rfc)));
			return;
		}
		// 如果是同步或异步冲正，则创建并启动反交易流程
		log.warn("Start flow reverser {} from {} for flow {}. RevType = {}", //
				rfc, re, flow, rt);
		SimpleFlow rtf = this.create(rfc);
		final SimpleFlow rfo = rtf.newInstance();
		rfo.setRevFor(flow.getId());
		rfo.setData(flow.getData());
		rfo.setRevTime(flow.getRevTime() + 1);
		rfo.setEntry(re);
		// 反交易start时，persister会将流程保存到数据库，此处不应显式保存
		// persister.save(rfo);
		// 更新冲正状态为正在冲正，冲正完成时，在更新反交易状态时会一并更新正交易的状态
		flow.setRevStatus(SimpleFlow.REV_EXECUTING);
		// flow.setRevTime(1);//冲正次数也是在完成时更新
		persister.update(flow);
		//
		switch (rt) {
		case SimpleFlow.REV_TYPE_ASYNC:
			this.execute(rfo);
			break;
		case SimpleFlow.REV_TYPE_SYNC:
			this.start(rfo);
			break;
		}
	}

	@Override
	public void execute(final SimpleFlow flow) {
		final SimpleFlowEngine fe = this;
		threadPool.execute(new Runnable() {
			@Override
			public void run() {
				try {
					fe.start(flow);
				} catch (Throwable e) {
					log.error("Error on execute flow {}", flow, e);
				}
			}
		});
	}

	@Override
	public void execute(String flowId, String id, Object params) {
		SimpleFlow f = this.create(flowId);
		f.setId(id);
		f.setData(params);
		this.execute(f);
	}

	@Override
	public void execute(String flowId, String id, Object params, String entry) {
		SimpleFlow f = this.create(flowId);
		f.setId(id);
		f.setData(params);
		if (Util.strIsNotEmpty(entry)) {
			f.setEntry(entry);
		}
		this.execute(f);
	}

	@Override
	public Future<SimpleFlow> submit(final SimpleFlow flow) {
		// 设置startTime以避免在流程引擎内部去计时
		flow.setStartTime(System.currentTimeMillis());
		final SimpleFlowEngine fe = this;
		Future<SimpleFlow> ft = threadPool.submit(new Callable<SimpleFlow>() {
			@Override
			public SimpleFlow call() throws Exception {
				fe.mark(flow);
				try {
					return fe.start(flow);
				} catch (Throwable e) {
					log.error("Error on submit flow {}", flow, e);
					throw new Exception(e);
				} finally {
					fe.unmark(flow);
				}
			}
		});
		return ft;
	}

	@Override
	public Future<SimpleFlow> submit(String flowId, String id, Object params, String entry) {
		SimpleFlow f = this.create(flowId);
		f.setId(id);
		f.setData(params);
		if (Util.strIsNotEmpty(entry)) {
			f.setEntry(entry);
		}
		return this.submit(f);
	}

	public BaseNode getNode(String flowId, String beanId) throws Exception {
		//
		Map<String, BaseNode> bm = nodeMap.get(flowId);
		if (bm == null) {
			synchronized (nodeMap) {
				bm = nodeMap.get(flowId);
				if (bm == null) {
					bm = new ConcurrentHashMap<String, BaseNode>();
					nodeMap.put(flowId, bm);
				}
			}
		}
		//
		BaseNode n = bm.get(beanId);
		if (n == null) {
			synchronized (bm) {
				n = bm.get(beanId);
				if (n == null) {
					//
					String bid = String.format("%s-%s", flowId, beanId);
					if (applicationContext.containsBean(bid)) {
						n = (BaseNode) applicationContext.getBean(bid);
						n.setId(bid);
					} else if (applicationContext.containsBean(beanId)) {
						n = (BaseNode) applicationContext.getBean(beanId);
						n.setId(beanId);
					} else {
						throw new IllegalStateException(String.format(//
								"There is no either %s or %s bean", bid, beanId));
					}
					bm.put(beanId, n);// 放入缓存
				}
			}
		}
		return n;
	}

	protected void suspend(SimpleFlow flow, SuspendNode suspendNode) throws Throwable {
		log.warn("Suspend flow {} at {}", flow, suspendNode.getId());
		flow.setFlag(SimpleFlow.FLAG_SUSPEND);
		if (suspendNode.getHandler() != null) {
			suspendNode.getHandler().save(flow);
		}
		this.persister.update(flow);
	}

	@Override
	public SimpleFlow resume(String flowId, String id, Object exData) throws Throwable {
		SimpleFlow flow = this.getFlow(flowId, id);
		return this.resume(flow, exData);
	}

	@Override
	public SimpleFlow resume(SimpleFlow flow) throws Throwable {
		return this.resume(flow, null);
	}

	@Override
	public SimpleFlow resume(SimpleFlow flow, Object data) throws Throwable {
		String flowId = flow.getFlowId();
		String step = flow.getStep();
		this.mark(flow);
		try {
			if (step.contains("/")) {
				log.info("Resume flow {} from {} and deadline is {}", flow, step,
						Util.date2str(new Date(flow.getDeadline()), Util.STDDTMF));
				// TODO 暂不支持并行分支中的子流程有挂起的情形
				String[] sa = step.split("/"); // flowId/id
				SimpleFlow subflow = this.getFlow(sa[0], sa[1]);
				SuspendNode sn = (SuspendNode) this.getNode(sa[0], subflow.getStep());
				// 续跑内嵌的subflow
				subflow.setFlag(SimpleFlow.FLAG_GO);
				Object ro = this.go(subflow, sn.getNext(), data, subflow.getError(), subflow.getTrace());
				subflow.setData(ro);
				// 如果这一subflow再次挂起，依旧挂起主流程
				if (subflow.getFlag() == SimpleFlow.FLAG_SUSPEND) {
					return flow;
				}
				// 如果内嵌的子流程跑完了，就继续主流程
				flow.setStep(sa[2]);
				// 使用父流程自己的数据来续跑，而不能用子流程的数据
				data = flow.getData();
			} else {
				SuspendNode sn = (SuspendNode) this.getNode(flowId, step);
				log.info("Resume flow {} from {} after {} and deadline is {}", flow, sn.getNext(), step,
						Util.date2str(new Date(flow.getDeadline()), Util.STDDTMF));
				if (sn.getHandler() != null) {
					sn.getHandler().merge(flow, data);
				}
				flow.setStep(sn.getNext());
			}
			flow.setFlag(SimpleFlow.FLAG_GO);
			data = this.go(flow, flow.getStep(), data, null, flow.getTrace());
			flow.setData(data);
		} finally {
			//
			if (flow.getFlag() != SimpleFlow.FLAG_SUSPEND) {
				log.debug("Remove resumed flow {} from context since it stopped", flow);
				this.removeFlow(flow);
			}
			this.onFinally(flow);
		}
		return flow;
	}

	protected boolean isEndStep(String step) {
		return "end".equalsIgnoreCase(step) || "null".equalsIgnoreCase(step) || Util.strIsEmpty(step);
	}

	/**
	 * 
	 * @param flow
	 * @param step
	 * @param data
	 *            flow的主数据已在data属性中，forkData为额外传递的分支数据
	 * @return
	 * @throws Throwable
	 */
	public Object go(final SimpleFlow flow, String step, Object data, Throwable flowEx, FlowTrace trace) throws Throwable {
		data = (data == null ? flow.getData() : data);
		while (flow.getFlag() == SimpleFlow.FLAG_GO) {
			if (this.isEndStep(step)) {
				flow.setFlag(SimpleFlow.FLAG_END);
				listener.onBeforeStep(flow, data, null);
				return data;
			}
			Throwable stepEx = null;
			BaseNode n = this.getNode(flow.getFlowId(), step);
			long sst = System.currentTimeMillis();
			listener.onBeforeStep(flow, data, n);
			flow.setStep(step);
			int status = n.getStatus();
			// 入栈当前步骤ID，并以此ID标记当前线程
			// log.info("Proceed flow {} from bean {}", flow, n.getId());
			// 登记调用链（实际是否登记取决于tracing开关和流程的采样率ratio）
			this.push(flow, n.getId(), data, trace);
			try {
				switch (n.getType()) {
				case BaseNode.TYPE_SUBFLOW:
					SubflowNode subflowNode = (SubflowNode) n;
					data = this.doSubflow(subflowNode, status, flow, data, flowEx, trace);
					step = subflowNode.getNext();
					continue;
				case BaseNode.TYPE_PROCESS:
					ProcessNode processNode = (ProcessNode) n;
					data = this.doProcess(processNode, status, flow, data, flowEx);
					step = processNode.getNext();
					continue;
				case BaseNode.TYPE_JUDGE:
					JudgeNode judgeNode = (JudgeNode) n;
					step = this.doJudge(judgeNode, status, flow, data, flowEx);
					continue;
				// case BaseNode.TYPE_START: //只有前台才有Start节点，后台没有
				// StartNode startNode = (StartNode) n;
				// beanId = startNode.getNext();
				// break;
				case BaseNode.TYPE_END:
					flow.setFlag(SimpleFlow.FLAG_END);
					return data;
				case BaseNode.TYPE_SWITCH:
					SwitchNode switchNode = (SwitchNode) n;
					step = this.doSwitch(switchNode, status, flow, data, flowEx);
					continue;
				case BaseNode.TYPE_INPUT:
					InputNode inputNode = (InputNode) n;
					if (status == BaseNode.STATUS_NORMAL) {
						this.suspend(flow, inputNode);
					} else {
						log.warn("Skip {} of flow {} because status is {} ", n.getId(), flow, status);
					}
					return new UiBean(inputNode.getUi(), data);
				// case BaseNode.TYPE_OUTPUT:
				// if (status == BaseNode.STATUS_NORMAL) {
				// this.suspend(flow, (SuspendNode) n);
				// } else {
				// log.warn("Skip {} of flow {} because status is {} ",
				// n.getId(), flow, status);
				// }
				// return ((OutputNode) n).getUi();
				case BaseNode.TYPE_TRY:
					TryNode tryNode = (TryNode) n;
					data = this.doTry(tryNode, status, flow, data, flowEx, trace);
					YrtNode yrtNode = (YrtNode) this.getNode(flow.getFlowId(), tryNode.getYrt());
					step = yrtNode.getNext();
					continue;
				case BaseNode.TYPE_CATCH: // try的返回标识
				case BaseNode.TYPE_FINALLY: // try或catch的返回标识
				case BaseNode.TYPE_YRT: // try或finally的返回标识
					return data; // 单纯返回，具体处理都在TRY中
				case BaseNode.TYPE_BEGIN_TX:
					BeginTxNode txNode = (BeginTxNode) n;
					data = this.doBeginTx(txNode, status, flow, data, flowEx, trace);
					CommitTxNode commitTxNode = (CommitTxNode) this.getNode(flow.getFlowId(), txNode.getCommitTx());
					data = this.go(flow, commitTxNode.getNext(), data, flowEx, trace);
					return data;
				case BaseNode.TYPE_COMMIT_TX:
				case BaseNode.TYPE_ROLLBACK_TX:
					return data; // 单纯返回
				case BaseNode.TYPE_FORK:
					ForkNode forkNode = (ForkNode) n;
					JoinNode jn = (JoinNode) this.getNode(flow.getFlowId(), forkNode.getJoin());
					if (status == BaseNode.STATUS_NORMAL) {
						data = this.doFork(flow, forkNode, jn, data, flowEx, trace);
					} else {
						log.warn("Skip {} of flow {} because status is {} ", n.getId(), flow, status);
					}
					step = jn.getNext();
					continue;
				case BaseNode.TYPE_JOIN:
					// 如果未经过fork节点（非递归），则继续向下，而不是返回
					// JoinNode joinNode = (JoinNode) n;
					// TODO 以JOIN为入口时的处理？
					return data;
				case BaseNode.TYPE_SUSPEND:
					if (status == BaseNode.STATUS_NORMAL) {
						this.suspend(flow, (SuspendNode) n);
					} else {
						log.warn("Skip {} of flow {} because status is {} ", n.getId(), flow, status);
					}
					return data;
				default:
					throw new IllegalStateException(String.format("Unknown node type %d", n.getType()));
				}
			} catch (Throwable e) {
				stepEx = e;
				listener.onFailStep(flow, data, flowEx, step, e);
				throw e;
			} finally {
				this.pop(flow, trace, data);
				listener.onAfterStep(flow, data, flowEx, step, stepEx, System.currentTimeMillis() - sst);
			}
			// } catch (Throwable e) {
			// log.error("Error on do node {}", n, e);
			// throw e;
		}
		return data;
	}

	protected String doJudge(JudgeNode judgeNode, int status, SimpleFlow flow, Object data, Throwable flowEx) throws Throwable {
		if (status != BaseNode.STATUS_NORMAL) {
			log.warn("Skip {} of flow {} because status is {} ", judgeNode.getId(), flow, status);
			return judgeNode.getYes();
		}
		WrapHandler wrapper = judgeNode.getWrapper();
		Object checkInData = wrapper.checkIn(judgeNode.getId(), data);
		Throwable handleEx = null;
		String rs = null;
		listener.onBeforeHandle(flow, data, flowEx, checkInData);
		try {
			rs = judgeNode.getNext(checkInData, flowEx);
			return rs;
		} catch (Throwable e) {
			handleEx = e;
			listener.onFailHandle(flow, data, flowEx, checkInData, handleEx);
			throw e;
		} finally {
			listener.onAfterHandle(flow, data, flowEx, checkInData, rs, handleEx);
		}
	}

	protected String doSwitch(SwitchNode switchNode, int status, SimpleFlow flow, Object data, Throwable flowEx)
			throws Throwable {
		if (status != BaseNode.STATUS_NORMAL) {
			log.warn("Skip {} of flow {} because status is {} ", switchNode.getId(), flow, status);
			return switchNode.getDefault();
		}
		WrapHandler wrapper = switchNode.getWrapper();
		Object checkInData = wrapper.checkIn(switchNode.getId(), data);
		String rs = null;
		Throwable handleEx = null;
		listener.onBeforeHandle(flow, data, flowEx, checkInData);
		try {
			rs = switchNode.getNext(checkInData, flowEx);
			return rs;
		} catch (Throwable e) {
			handleEx = e;
			listener.onFailHandle(flow, data, flowEx, checkInData, handleEx);
			throw e;
		} finally {
			listener.onAfterHandle(flow, data, flowEx, checkInData, rs, handleEx);
		}
	}

	protected Object doBeginTx(BeginTxNode txNode, int status, SimpleFlow flow, Object data, Throwable flowEx, FlowTrace trace)
			throws Throwable {
		if (status != BaseNode.STATUS_NORMAL) {
			log.warn("Skip {} of flow {} because status is {} ", txNode.getId(), flow, status);
			return data;
		}
		TransactionHandler<Object> txHandler = txNode.getHandler();
		Object tx = txHandler.begin(data);
		try {
			// 向下执行，直到遇到commit或rollback节点才返回
			Object ro = _this.go(flow, txNode.getNext(), data, flowEx, trace);
			txHandler.commit(tx);
			return ro;
		} catch (Throwable e) {
			txHandler.rollback(tx);
			throw e;
		}
	}

	protected Object doSubflow(SubflowNode subflowNode, int status, SimpleFlow flow, Object data, Throwable flowEx,
			FlowTrace trace) throws Throwable {
		if (status != BaseNode.STATUS_NORMAL) {
			log.warn("Skip {} of flow {} because status is {} ", subflowNode.getId(), flow, status);
			return data;
		}
		//
		WrapHandler wrapper = subflowNode.getWrapper();
		Object checkInData = wrapper.checkIn(flow.getFlowId(), data);
		Object dataCheckOut = null;
		Throwable handleEx = null;
		listener.onBeforeHandle(flow, data, flowEx, checkInData);
		try {
			//
			SimpleFlow subflow = _this.create(subflowNode.getRef());
			subflow.setParent(flow.getId());
			subflow.setData(checkInData);
			_this.start(subflow);
			//
			dataCheckOut = subflow.getData();
			if (subflow.getFlag() == SimpleFlow.FLAG_SUSPEND) {
				flow.setFlag(SimpleFlow.FLAG_SUSPEND);
				flow.setStep(new StringBuilder(subflow.getFlowId())//
						.append("/").append(subflow.getId()).append("/")//
						.append(subflowNode.getNext()).toString());
			}
			// 将子流程的trace合并过来
			_this.merge(flow, trace, subflow.getTrace());
		} catch (Throwable e) {
			handleEx = e;
			listener.onFailHandle(flow, data, flowEx, checkInData, handleEx);
			throw e;
		} finally {
			listener.onAfterHandle(flow, data, flowEx, checkInData, dataCheckOut, handleEx);
		}
		return wrapper.checkOut(dataCheckOut, data);
	}

	protected Object doProcess(ProcessNode processNode, int status, SimpleFlow flow, Object data, Throwable flowEx)
			throws Throwable {
		if (status != BaseNode.STATUS_NORMAL) {
			log.warn("Skip {} of flow {} because status is {} ", processNode.getId(), flow, status);
			return data;
		}
		String entry = processNode.getEntry();
		WrapHandler wrapper = processNode.getWrapper();
		Object checkInData = wrapper.checkIn(entry, data);
		Object dataCheckOut = null;
		Throwable handleEx = null;
		listener.onBeforeHandle(flow, data, flowEx, checkInData);
		try {
			dataCheckOut = processNode.getHandler().process(entry, checkInData, flowEx);
		} catch (Throwable e) {
			handleEx = e;
			listener.onFailHandle(flow, data, flowEx, checkInData, handleEx);
			throw e;
		} finally {
			listener.onAfterHandle(flow, data, flowEx, checkInData, dataCheckOut, handleEx);
		}
		return wrapper.checkOut(dataCheckOut, data);
	}

	protected Object doTry(TryNode tryNode, int status, SimpleFlow flow, Object data, Throwable flowEx, FlowTrace trace)
			throws Throwable {
		if (status != BaseNode.STATUS_NORMAL) {
			log.warn("Skip {} of flow {} because status is {} ", tryNode.getId(), flow, status);
			return data;
		}
		try {
			return this.go(flow, tryNode.getNext(), data, flowEx, trace);
		} catch (Throwable e) {
			if (tryNode.getCatch() == null) {
				log.error("Error on try {} but no catch node", tryNode.getId(), e);
				throw e; // 如果无finally块，此句才会生效，中断当前流程
			}
			CatchNode catchNode = (CatchNode) this.getNode(flow.getFlowId(), tryNode.getCatch());
			if (catchNode.getStatus() == BaseNode.STATUS_DISABLED) {
				log.error("Error on try {}, but cacth {} is disabled", //
						tryNode.getId(), tryNode.getCatch(), e);
				throw e;
			}
			// try catch 嵌套时，不把传入外层的异常传入内层，内层代码只关心内层的异常
			// flow.setError(flowEx);
			return this.go(flow, catchNode.getNext(), data, e, trace);
		} finally {
			if (tryNode.getFinally() != null) {
				short f0 = flow.getFlag();
				try {
					// 流程停止或出异常中断时，要保证try...finally的finally一定能得以执行
					flow.setFlag(SimpleFlow.FLAG_GO);
					FinallyNode finallyNode = (FinallyNode) this.getNode(flow.getFlowId(), tryNode.getFinally());
					if (finallyNode.getStatus() == BaseNode.STATUS_NORMAL) {
						return this.go(flow, finallyNode.getNext(), data, flowEx, trace);
					}
				} finally {
					// finally 执行完成后恢复回原来的flag
					flow.setFlag(f0);
				}
			}
		}
	}

	protected Object doFork(final SimpleFlow flow, ForkNode forkNode, JoinNode joinNode, Object data, final Throwable flowEx,
			final FlowTrace trace) throws Throwable {
		Object checkInData = forkNode.getWrapper().checkIn(forkNode.getId(), data);
		Object dataCheckOut = null;
		Throwable handleEx = null;
		listener.onBeforeHandle(flow, data, flowEx, checkInData);
		try {
			Map<String, Future<Object>> futureMap = new HashMap<String, Future<Object>>();
			final Map<String, Object> resultMap = new HashMap<String, Object>();
			final SimpleFlowEngineImpl fe = this;
			Set<String> branchs = forkNode.getBranches();
			// 如果有配handler，则遍历所有branch，用其key作为参数，调用handler的fork方法返回多个要并行的分支及其数据
			if (forkNode.getHandler() == null) {
				for (final String br : branchs) {
					Future<Object> ft = forkNode.getThreadPool().submit(new Callable<Object>() {
						//
						@Override
						public Object call() throws Exception {
							FlowTrace t = new SimpleFlowTrace(flow.getFlowId(), flow.getId());
							fe.mark(flow);
							try {
								return fe.go(flow, br, flow.getData(), flowEx, t);
							} catch (Throwable e) {
								log.error("Error on do fork branch {}", br, e);
								throw new Exception(e);
								// return e;
							} finally {
								listener.onAfterBranch(flow);
								fe.unmark(flow);
								fe.merge(flow, trace, t);
							}
						}
					});
					futureMap.put(br, ft);
				}
			} else {
				// 反之，遍历所有branch直接向下递归
				for (final String br : branchs) {
					final Map<String, Object> forks = forkNode.getHandler().fork(br, checkInData, flowEx);
					if (forks.isEmpty()) {
						continue;
					}
					log.debug("Fork branch {} has {} forkings", br, forks.size());
					for (final Entry<String, Object> en : forks.entrySet()) {
						String k = en.getKey();
						final Object d = en.getValue();
						Future<Object> ft = forkNode.getThreadPool().submit(new Callable<Object>() {
							//
							@Override
							public Object call() throws Exception {
								FlowTrace t = new SimpleFlowTrace(flow.getFlowId(), flow.getId());
								fe.mark(flow);
								try {
									return fe.go(flow, br, d, flowEx, t);
								} catch (Throwable e) {
									log.error("Error on do fork branch {}", br, e);
									throw new RuntimeException(e);
									// return e;
								} finally {
									listener.onAfterBranch(flow);
									fe.unmark(flow);
									fe.merge(flow, trace, t);
								}
							}

							// 实现toString以便于识别分支
							@Override
							public String toString() {
								return new StringBuilder(br).append("/").append(d).toString();
							}
						});

						// 这 里要求 handler.fork() 返回的KEY必须是唯一的
						String fk = Util.strIsEmpty(br) ? k : new StringBuilder(br).append("-").append(k).toString();
						if (futureMap.containsKey(k)) {
							throw new IllegalStateException("Duplicated fork key ".concat(fk));
						}
						futureMap.put(fk, ft);
					}
				}
			}
			//
			for (Entry<String, Future<Object>> en : futureMap.entrySet()) {
				String key = en.getKey();
				log.debug("Put future {} into future map", key);
				Future<?> ft = en.getValue();
				Object o = ft.get();
				if (o != null) {
					if (o instanceof Throwable) {
						Throwable e = (Throwable) o;
						log.error("Fork branch {}/{} throw exception {}", //
								forkNode.getId(), key, e.getMessage(), e);
					}
					resultMap.put(key, o);
				}
			}
			// 待前面future都结束，才判断是否有异常
			for (Entry<String, Future<Object>> en : futureMap.entrySet()) {
				Future<?> ft = en.getValue();
				Object o = ft.get();
				if (o != null && o instanceof Throwable) {
					throw (Throwable) o;
				}
			}
			log.debug("Join node {} join {} futures", joinNode, resultMap.size());
			dataCheckOut = joinNode.join(data, resultMap);
		} catch (Throwable e) {
			handleEx = e;
			listener.onFailHandle(flow, data, flowEx, checkInData, e);
			throw e;
		} finally {
			listener.onAfterHandle(flow, data, flowEx, checkInData, dataCheckOut, handleEx);
		}
		return joinNode.getWrapper().checkOut(checkInData, data);
	}

	protected void push(SimpleFlow flow, String step, Object data, FlowTrace trace) {
		if (tracing && flow.isTracing()) {
			try {
				Object d = flow.isTraceIo() ? this.clone(data) : null;
				trace.push(step, d);
			} catch (ClassNotFoundException | IOException e) {
				log.error("Can not push trace for {}", flow, e);
			}
		}
	}

	protected void pop(SimpleFlow flow, FlowTrace trace, Object data) {
		if (tracing && flow.isTracing()) {
			try {
				Object d = flow.isTraceIo() ? this.clone(data) : null;
				trace.pop(d);
			} catch (ClassNotFoundException | IOException e) {
				log.error("Can not push trace for {}", this, e);
			}
		}
	}

	protected void merge(SimpleFlow flow, FlowTrace trace1, FlowTrace trace2) {
		if (tracing && flow.isTracing()) {
			trace1.merge(trace2);
		}
	}

	public boolean isTracing() {
		return tracing;
	}

	public void setTracing(boolean tracing) {
		this.tracing = tracing;
	}

	protected Object clone(Object data) throws IOException, ClassNotFoundException {
		byte[] b = serializer.serialize(data);
		return serializer.deserialize(b);
	}

	@Override
	public void close() {
		this.nodeMap.clear();
		this.defsMap.clear();
		if (svcSpy == null) {
			return;
		}
		svcSpy.getGlobalSpyData().getSvcConfig().setStatus(SvcConfig.STATUS_DISABLE);
	}

	@Override
	public void open() {
		if (svcSpy == null) {
			return;
		}
		svcSpy.getGlobalSpyData().getSvcConfig().setStatus(SvcConfig.STATUS_ENABLE);
	}

	@Override
	public void mark(SimpleFlow flow) {
		SimpleFlow sf = SimpleFlowContext.getCurrentFlow();
		if (sf != null) {
			flow.setParent(sf.getId());
		}
		// 此方法通过push堆栈解决流程嵌套问题
		SimpleFlowContext.setCurrentFlow(flow);
		// 此动作会覆盖RPC维护的traceId，导致RPC调用链日志中断
		MDC.put(FlowConfig.TRACE_TAG, flow.getId());
		// 需在调用链查询时用flow.trace.parent关联
		tracker.track(flow);
	}

	@Override
	public void unmark(SimpleFlow flow) {
		SimpleFlowContext.removeCurrentFlow();
		MDC.remove(FlowConfig.TRACE_TAG);
		tracker.untrack(flow);
		SimpleFlow sf = SimpleFlowContext.getCurrentFlow();
		if (sf != null) {
			this.mark(sf);
		}
	}

	@Override
	public Map<String, Map<String, SimpleFlow>> getTimeMap() {
		return timeMap;
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.applicationContext = applicationContext;
	}

	public ApplicationContext getApplicationContext() {
		return applicationContext;
	}

	public ExecutorService getThreadPool() {
		return threadPool;
	}

	public void setThreadPool(ExecutorService threadPool) {
		this.threadPool = threadPool;
	}

	@Override
	public SimpleFlowPersister getPersister() {
		return persister;
	}

	public void setPersister(SimpleFlowPersister persister) {
		this.persister = persister;
	}

	public SimpleFlowListener getListener() {
		return listener;
	}

	public void setListener(SimpleFlowListener listener) {
		this.listener = listener;
	}

	public ScheduledExecutorService getScheduler() {
		return scheduler;
	}

	public void setScheduler(ScheduledExecutorService scheduler) {
		this.scheduler = scheduler;
	}

	@Override
	public String getCatalog() {
		return catalog;
	}

	public void setCatalog(String catalog) {
		this.catalog = catalog;
	}

	public ObjectOutput getMessageSender() {
		return messageSender;
	}

	public void setMessageSender(ObjectOutput messageSender) {
		this.messageSender = messageSender;
	}

	public SimpleFlowTracker getTracker() {
		return tracker;
	}

	public void setTracker(SimpleFlowTracker tracker) {
		this.tracker = tracker;
	}

}